package com.dtruth.kafka.producer;

import java.io.IOException;
import java.util.Properties;

import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;

import com.dtruth.kafka.message.User;
import com.feheadline.monitor.util.Serializer;

/**
 * 
 * @author 岳海亮
 * @email yhl@feheadline.com
 * @date 2014年11月12日
 */
public class KafkaProducer extends Thread {
	private final kafka.javaapi.producer.Producer<Integer, byte[]> producer;
	private final String topic;
	private final Properties props = new Properties();

	public KafkaProducer(String topic) {
		// props.put("serializer.class", "kafka.serializer.StringEncoder");
		props.put("serializer.class", "kafka.serializer.DefaultEncoder");
		props.put("metadata.broker.list", "tree:9092");
		props.put("num.partitions", 2);
		producer = new kafka.javaapi.producer.Producer<Integer, byte[]>(
				new ProducerConfig(props));
		this.topic = topic;
	}

	@Override
	public void run() {
		int messageNo = 1;
		try {
			while (true) {
				User user = new User();
				user.setId(messageNo);
				user.setName("name-" + messageNo);
				System.out.println("Send:" + messageNo);
				producer.send(new KeyedMessage<Integer, byte[]>(topic, Serializer.serialize(user)));
				messageNo++;
				try {
					sleep(3000);
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			}
		} catch (IOException e) {
			e.printStackTrace();
		}
	}
}